package com.gitee.ixtf.cqrs.verticle

import com.gitee.ixtf.cqrs.kotlinx.ComponentConsumerMessage
import com.gitee.ixtf.cqrs.kotlinx.EB_ADDRESS_EVENT_SOURCED_ENTITY_CONSUMER
import com.gitee.ixtf.vertx.verticle.BaseCoroutineVerticle
import io.vertx.core.eventbus.MessageConsumer
import io.vertx.kotlin.coroutines.awaitEvent
import io.vertx.kotlin.coroutines.toReceiveChannel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch

abstract class EventSourcedEntityConsumer : BaseCoroutineVerticle() {
  protected val consumer: MessageConsumer<ComponentConsumerMessage<*, *>> by lazy {
    eventBus.consumer(EB_ADDRESS_EVENT_SOURCED_ENTITY_CONSUMER)
  }

  override suspend fun start() {
    super.start()
    launch {
      consumer.toReceiveChannel(vertx).consumeEach { reply ->
        runCatching { handle(reply.body()) }
            .onFailure { log.error(it, "address: ${consumer.address()} \n body: ${reply.body()}") }
      }
    }
    awaitEvent { consumer.completionHandler(it) }
  }

  override suspend fun stop() {
    consumer.unregister()
    super.stop()
  }

  protected abstract fun handle(message: ComponentConsumerMessage<*, *>)
}
